package com.alisoft.nano.seda.impl;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.management.InstanceNotFoundException;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;

import org.springframework.jmx.export.annotation.AnnotationMBeanExporter;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedNotification;
import org.springframework.jmx.export.annotation.ManagedNotifications;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.jmx.export.naming.SelfNaming;
import org.springframework.jmx.export.notification.NotificationPublisher;
import org.springframework.jmx.export.notification.NotificationPublisherAware;
import org.springframework.jmx.support.JmxUtils;

import com.alisoft.nano.seda.Event;
import com.alisoft.nano.seda.EventHandler;
import com.alisoft.nano.seda.Stage;
import com.alisoft.nano.seda.StageRuntimeException;
import com.alisoft.nano.seda.monitor.StageMonitor;
import com.alisoft.nano.seda.notification.StageNotification;
import com.alisoft.nano.seda.notification.StageNotificationListener;

/**
 * The spring implementation of the Stage
 * 
 * 
 * @author zhimin.chenzm, anping.yinap
 * 
 * @param <T>
 */
@ManagedResource(description = "Default Work Stage")
@ManagedNotifications(@ManagedNotification(name = "Stage Notification", notificationTypes = { StageNotification.PREPOST,
        StageNotification.ACCEPTED, StageNotification.PRERUN, StageNotification.DONE }))
public class DefaultStage<T> implements Stage<T>, NotificationPublisherAware, SelfNaming {

    private static final int DELAY = 50;

    private String mBeanDomain = "com.alisoft.nano.seda";
    private String mBeanName;

    private StageMonitor<T> stageMonitor;

    private List<StageNotificationListener<T>> listeners = new ArrayList<StageNotificationListener<T>>();

    /**
     * The pending event queue. Initialized in the constructor.
     * 
     */
    private final BlockingQueue<Event<T>> pendingEventQueue;

    /**
     * The name of the stage
     */
    private final String name;

    /**
     * The event handler to handle event Can be set in the setter.
     * 
     * Can't be null when the stage is started.
     */
    private EventHandler<T> eventHandler;

    /**
     * The executor service to execute the worker job at scheduled rate.
     * 
     * It is initialized in the start method. And destroyed in the stop method.
     */
    private ScheduledExecutorService workerExecutorService;

    /**
     * The executor service to process the work unit.
     * 
     * It is initialized in the start method. And destroyed in the stop method.
     */
    private ThreadPoolExecutor workUnitExecutorService;

    /**
     * The maximum size of thread pool in workUnitExecutorService.
     * 
     * Default is 10
     */
    private int maximumPoolSize = 10;

    /**
     * The size of the batch events which the worker polls from the event queue
     * and process them in the thread pool. Default is 10
     */
    private int batchSize = 10;

    /**
     * The default timeout to wait when posting a new event to this stage. (in
     * seconds) The default value is 10 minutes.
     */
    private long defaultTimeout = 10 * 60;

    /**
     * The publisher to publish notifications Injected by the spring container.
     */
    private NotificationPublisher publisher;

    /**
     * Indicate whether this stage is started.
     */
    private volatile boolean started = false;

    public DefaultStage(String name) {
        this(name, new LinkedBlockingQueue<Event<T>>());
    }

    public DefaultStage(String name, BlockingQueue<Event<T>> pendingEventQueue) {
        super();
        this.name = name;
        this.mBeanName = name;
        this.pendingEventQueue = pendingEventQueue;
    }

    /**
     * Post the event with default timeout.
     */
    public boolean post(Event<T> event) throws InterruptedException {
        return this.post(event, this.defaultTimeout, TimeUnit.SECONDS);
    }

    /**
     * Post the event with specified timeout to the pendingEventQueue Before
     * adding to queue, it will publish "newPrePost" notification. After added
     * to the queue, it will publish "Accepted" notification.
     * 
     * @param event
     * @param timeout
     * @param unit
     * @return
     * @throws InterruptedException
     */
    private boolean post(Event<T> event, long timeout, TimeUnit unit) throws InterruptedException {
        if (!this.isStarted()) {
            throw new IllegalStateException(this + " is not started");
        }

        this.publisher.sendNotification(StageNotification.newPrePost(event));
        boolean success = pendingEventQueue.offer(event, timeout, unit);
        if (success) {
            reset(event);
        }
        this.publisher.sendNotification(StageNotification.newAccepted(event));
        return success;
    }

    private void reset(Event<T> event) {
        event.setCurrentStage(this.getName());
        event.setException(null);
    }

    public String getName() {
        return this.name;
    }

    /**
     * Start this stage. If the stage is already started, do nothing If event
     * handler is not set yet, throw ISE If started successfully, publish the
     * "Started" notification
     */
    public synchronized void start() {
        if (this.started) {
            // already started
            return;
        }

        if (this.eventHandler == null) {
            throw new IllegalStateException("The EventHandler is missing");
        }

        
        registerJMX();
        

        this.workUnitExecutorService = new ThreadPoolExecutor(0, this.maximumPoolSize, 60L, TimeUnit.SECONDS,
                new SynchronousQueue<Runnable>(), Executors.defaultThreadFactory());

        this.workerExecutorService = Executors.newSingleThreadScheduledExecutor();
        this.workerExecutorService.scheduleWithFixedDelay(new Worker(), 0, DELAY, TimeUnit.MILLISECONDS);

        this.started = true;

        StageNotification<T> notification = StageNotification.newStarted(this);
        this.publisher.sendNotification(notification);

    }

    private void registerJMX() {

        // register JMX 
        AnnotationMBeanExporter mBeanExporter = new AnnotationMBeanExporter();
        MBeanServer server = JmxUtils.locateMBeanServer();
        mBeanExporter.setServer(server);

        // register this stage and monitorMBean
        ObjectName name;
        try {
            name = this.getObjectName();
        } catch (MalformedObjectNameException e) {
            throw new StageRuntimeException(this.mBeanDomain + ":name=" + this.mBeanName + " is not a valid MBean object name", e);
        }

        mBeanExporter.registerManagedResource(this, name);
        if (this.stageMonitor != null) {
            try {
                ObjectName monitorMBeanName = new ObjectName(this.mBeanDomain + ":name=" + this.mBeanName + "-monitor");
                mBeanExporter.registerManagedResource(this.stageMonitor, monitorMBeanName);
            } catch (MalformedObjectNameException e) {
                throw new StageRuntimeException(this.mBeanDomain + ":name=" + this.mBeanName + "-monitor"
                        + " is not a valid MBean object name", e);
            }
        }
        // register notification listeners
        try {
            if (this.stageMonitor != null) {
                server.addNotificationListener(name, stageMonitor, stageMonitor, null);
            }
            for (StageNotificationListener<T> listener : this.listeners) {
                server.addNotificationListener(name, listener, listener, null);
            }

        } catch (InstanceNotFoundException e) {
            throw new StageRuntimeException("Can't register notification listeners", e);
        }

    }

    /**
     * Stop the stage. (Problems in Current version - the pending event in the
     * queue will be discarded May be we can wait until all the events are
     * processed. )
     */
    public synchronized void stop() {
        this.started = false;
        if (this.workerExecutorService != null) {
            this.workerExecutorService.shutdown();
        }
        if (this.workUnitExecutorService != null) {
            this.workUnitExecutorService.shutdown();
        }
        this.publisher.sendNotification(StageNotification.newStopped(this));
    }

    public void setEventHandler(EventHandler<T> eventHandler) {
        this.eventHandler = new EventHandlerWrapper<T>(eventHandler);
    }

    /**
     * The runner to run the batched events. This runner will be executed in the
     * workUnitExecutorService
     * 
     */
    private class EventHandlerRunner implements Runnable {

        private List<Event<T>> events;

        public EventHandlerRunner(List<Event<T>> events) {
            super();
            this.events = events;
        }

        public void run() {
            for (Event<T> event : events) {
                publisher.sendNotification(StageNotification.newPreRun(event));
                try {
                    eventHandler.process(event);
                } catch (Throwable e) {
                    // TODO: log it.
                }
                publisher.sendNotification(StageNotification.newDone(event));
            }
        }

    }

    /**
     * The worker job to poll events from pendingEventQueue and execute them on
     * workUnitExecutorService
     * 
     * 
     */
    private class Worker implements Runnable {

        public void run() {

            try {
                if (workUnitExecutorService.getActiveCount() == workUnitExecutorService.getMaximumPoolSize()) {
                    return;
                }
                List<Event<T>> events = new ArrayList<Event<T>>();
                for (int i = 0; i < batchSize; i++) {
                    Event<T> event = pendingEventQueue.poll(0, TimeUnit.MILLISECONDS);
                    if (event != null) {
                        events.add(event);
                    }
                }

                if (!events.isEmpty()) {
                    workUnitExecutorService.execute(new EventHandlerRunner(events));
                }

            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

    }

    @Override
    public String toString() {
        return "Stage[" + name + "]";
    }

    public boolean isStarted() {
        return this.started;
    }

    @ManagedAttribute
    public int getPendingEventCount() {
        return this.pendingEventQueue.size();
    }

    @ManagedAttribute
    public int getMaximumPoolSize() {
        if (this.isStarted()) {
            return this.workUnitExecutorService.getMaximumPoolSize();
        }
        return this.maximumPoolSize;
    }

    @ManagedAttribute
    public void setMaximumPoolSize(int maxSize) {
        this.maximumPoolSize = maxSize;
        if (this.isStarted()) {
            this.workUnitExecutorService.setMaximumPoolSize(maxSize);
        }

    }

    public void setNotificationPublisher(NotificationPublisher publisher) {
        this.publisher = publisher;
    }

    @ManagedAttribute
    public void setBatchSize(int batchSize) {
        this.batchSize = batchSize;
    }

    @ManagedAttribute
    public int getBatchSize() {
        return batchSize;
    }

    @ManagedAttribute
    public void setDefaultTimeout(long defaultTimeout) {
        this.defaultTimeout = defaultTimeout;
    }

    @ManagedAttribute
    public long getDefaultTimeout() {
        return defaultTimeout;
    }

    @ManagedAttribute
    public int getActiveThreads() {
        if (this.isStarted()) {
            return this.workUnitExecutorService.getActiveCount();
        }
        return 0;
    }

    public String getMBeanName() {
        return mBeanName;
    }

    public void setMBeanName(String beanName) {
        mBeanName = beanName;
    }

    public String getMBeanDomain() {
        return mBeanDomain;
    }

    public void setMBeanDomain(String beanDomain) {
        mBeanDomain = beanDomain;
    }

    public void addStageNotificationListener(StageNotificationListener<T> listener) {
        this.listeners.add(listener);
    }
    
    public void setListeners(List<StageNotificationListener<T>> listeners) {
        this.listeners = listeners;
    }

    public StageMonitor<T> getStageMonitor() {
        return this.stageMonitor;
    }

    public void setStageMonitor(StageMonitor<T> stageMonitor) {
        this.stageMonitor = stageMonitor;
    }


    public ObjectName getObjectName() throws MalformedObjectNameException {
        return new ObjectName(this.mBeanDomain + ":name=" + this.mBeanName);
    }

    
}
